package com.atbeijing.D05;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * 模拟5s滚动窗口,求每个窗口总温度
 */
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .filter(r -> r.id.equals("sensor_1"))
                .assignTimestampsAndWatermarks(
                    WatermarkStrategy
                            .<SensorReading>forMonotonousTimestamps()
                            .withTimestampAssigner(new SerializableTimestampAssigner<SensorReading>(){
                                @Override
                                public long extractTimestamp(SensorReading element, long recordTimestamp) {
                                    return element.timestamp;
                                }
                            })
                )
                .keyBy(r -> r.id)
                .process( new MyKeyedProcessFunction2())
                .print();

        env.execute();

    }

    public static class MyKeyedProcessFunction2 extends KeyedProcessFunction<String,SensorReading,String>{
        //状态变量,key存窗口结束时间,value存温度总和
        private MapState<Long,Double> tempState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            //初始化状态变量
            tempState=getRuntimeContext().getMapState(new MapStateDescriptor<Long, Double>("temp-State", Types.LONG,Types.DOUBLE));
        }

        @Override
        public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
            //同一个窗口的数据过来时窗口开始时间和结束时间不变
            //窗口开始时间戳,窗口长度为5s
            long windowStart=value.timestamp-value.timestamp%5000L;
            //伪窗口结束时间戳,因为窗口是左闭右开的所以收集数据的结束时间要减少1ms
            long windowEnd = windowStart + 5000L-1L;
            if (tempState.contains(windowEnd)){
                //累加温度
                tempState.put(windowEnd,tempState.get(windowEnd)+value.temperature);
            }else {
                tempState.put(windowEnd,value.temperature);
            }
            //时间戳相同的定时器只会注册一次,逻辑时钟定时器
            ctx.timerService().registerEventTimeTimer(windowEnd);
        }

        //timestamp=windowEnd时执行方法
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            //真正的窗口时间=windowEnd+1=timestamp+1
            out.collect(new Timestamp(timestamp+1L-5000L) +"~"+new Timestamp(timestamp+1L)+"\t"+tempState.get(timestamp));
        }
    }






}
